package com.gph.saviorframework.mq.api.impl.rocketmq;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.LocalTransactionState;
import com.alibaba.rocketmq.client.producer.TransactionCheckListener;
import com.alibaba.rocketmq.client.producer.TransactionMQProducer;
import com.alibaba.rocketmq.common.message.MessageAccessor;
import com.gph.saviorframework.mq.api.Constants;
import com.gph.saviorframework.mq.api.Message;
import com.gph.saviorframework.mq.api.PropertyKeyConst;
import com.gph.saviorframework.mq.api.SendResult;
import com.gph.saviorframework.mq.api.impl.util.MQUtil;
import com.gph.saviorframework.mq.api.transaction.LocalTransactionExecuter;
import com.gph.saviorframework.mq.api.transaction.TransactionProducer;
import com.gph.saviorframework.mq.api.transaction.TransactionStatus;

import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by root on 16-7-11.
 */
public class TransactionProducerImpl extends MQClientAbstract implements TransactionProducer {

    private final AtomicBoolean started = new AtomicBoolean(false);
    TransactionMQProducer transactionMQProducer = null;
    private Properties properties;

    public TransactionProducerImpl(Properties properties,TransactionCheckListener transactionCheckListener) {
        super(properties);
        this.properties = properties;
        transactionMQProducer =
                new TransactionMQProducer((String) properties.get(PropertyKeyConst.ProducerId), new ClientRPCHook(sessionCredentials));
        boolean isVipChannelEnabled = Boolean.parseBoolean(properties.getProperty(PropertyKeyConst.isVipChannelEnabled, "false"));
        transactionMQProducer.setVipChannelEnabled(isVipChannelEnabled);

        transactionMQProducer.setTransactionCheckListener(transactionCheckListener);
    }

    /**
     * 启动服务
     */
    public void start() {
        if (started.compareAndSet(false, true)) {
            if (transactionMQProducer.getTransactionCheckListener() == null) {
                throw new IllegalArgumentException("TransactionCheckListener is null");
            }
            // TODO 完善寻址功能。要实现TransactionMQProducer.sendMessageInTransaction
            transactionMQProducer.setNamesrvAddr(this.nameServerAddr);
            try {
                transactionMQProducer.start();
            } catch (MQClientException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 关闭服务
     */
    public void shutdown() {
        if (started.compareAndSet(true, false)) {
            transactionMQProducer.shutdown();
        }
    }

    /**
     * 发送消息
     *
     * @param message
     * @param executer 本地事务执行器
     * @param arg      应用自定义参数
     * @return
     */
    public SendResult send(final Message message, final LocalTransactionExecuter executer, Object arg) {
        this.checkProducerServiceState(this.transactionMQProducer.getDefaultMQProducerImpl());
        com.alibaba.rocketmq.common.message.Message msgRMQ = MQUtil.msgConvert(message);

        MessageAccessor.putProperty(msgRMQ, PropertyKeyConst.ProducerId, (String) properties.get(PropertyKeyConst.ProducerId));

        com.alibaba.rocketmq.client.producer.TransactionSendResult sendResultRMQ = null;
        try {
            sendResultRMQ = transactionMQProducer.sendMessageInTransaction(msgRMQ,
                    new com.alibaba.rocketmq.client.producer.LocalTransactionExecuter() {
                        public LocalTransactionState executeLocalTransactionBranch(com.alibaba.rocketmq.common.message.Message msg, Object arg) {
                            String msgId = msg.getProperty(Constants.TRANSACTION_ID);
                            message.setMsgID(msgId);
                            TransactionStatus transactionStatus = executer.execute(message, arg);
                            if (TransactionStatus.CommitTransaction == transactionStatus) {
                                return LocalTransactionState.COMMIT_MESSAGE;
                            }
                            else if (TransactionStatus.RollbackTransaction == transactionStatus) {
                                return LocalTransactionState.ROLLBACK_MESSAGE;
                            }
                            return LocalTransactionState.UNKNOW;
                        }
                    }, arg);
        }
        catch (Exception e) {
            throw new RuntimeException(e);
        }
        if (sendResultRMQ.getLocalTransactionState() == LocalTransactionState.ROLLBACK_MESSAGE) {
            // 本地事务显式说明了有异常，此时给应用方返回一个异常。
            throw new RuntimeException("local transaction branch failed ,so transaction rollback");
        }
        SendResult sendResult = new SendResult();
        sendResult.setMessageId(sendResultRMQ.getMsgId());
        if (sendResultRMQ.getTransactionId() != null) {
            sendResult.setMessageId(sendResultRMQ.getTransactionId());
            // sendResult.setTransactionId(sendResultRMQ.getTransactionId());
        }
        return sendResult;
    }

    public boolean isStarted() {
        return started.get();
    }

    public boolean isClosed() {
        return !isStarted();
    }
}
